package org.jetlinks.pro.device.message;

import lombok.AllArgsConstructor;
import org.jetlinks.core.device.DeviceConfigKey;
import org.jetlinks.core.device.DeviceOperator;
import org.jetlinks.core.message.DeviceMessage;
import org.jetlinks.core.message.Headers;
import org.jetlinks.core.message.ThingMessage;
import org.jetlinks.core.message.function.ThingFunctionInvokeMessage;
import org.jetlinks.core.message.interceptor.DeviceMessageSenderInterceptor;
import org.jetlinks.core.things.*;
import org.jetlinks.pro.things.ThingsConnector;
import org.jetlinks.pro.things.ThingsConnectorManager;
import org.springframework.core.Ordered;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.function.Function;

@AllArgsConstructor
@Component
public class ConnectedDeviceMessageSenderInterceptor implements DeviceMessageSenderInterceptor {

    private final ThingsConnectorManager connectorManager;

    private MetadataId getMetadataId(ThingMessage message) {
        if (message instanceof ThingFunctionInvokeMessage) {
            String functionId = ((ThingFunctionInvokeMessage<?>) message).getFunctionId();
            if (null == functionId) {
                return null;
            }
            return MetadataId.of(ThingMetadataType.function, functionId);
        }
        return null;
    }

    @Override
    public Flux<DeviceMessage> doSend(DeviceOperator device, DeviceMessage message, Flux<DeviceMessage> sender) {

        MetadataId metadataId = getMetadataId(message);
        if (metadataId == null) {
            return sender;
        }
        ThingId thingId = ThingId.of(message.getThingType(), message.getThingId());

        //获取连接器，如果支持rpc则使用rpc进行调用
        return connectorManager
            .getConnector(thingId, metadataId)
            //设备没有尝试获取产品的
            .switchIfEmpty(getProductRpcSupport(device, message, metadataId))

            .filter(connector -> connector.isWrapperFor(ThingRpcSupport.class))
            .map(connector -> connector
                .unwrap(ThingRpcSupport.class)
                .call(message)
                .cast(DeviceMessage.class)
            )
            .defaultIfEmpty(sender)
            .flatMapMany(Function.identity());
    }

    private Mono<ThingsConnector> getProductRpcSupport(DeviceOperator device, DeviceMessage message, MetadataId metadataId) {
        return Mono
            .justOrEmpty(message.getHeader(Headers.productId))
            .switchIfEmpty(device.getSelfConfig(DeviceConfigKey.productId))
            .flatMap(productId -> connectorManager.getConnector(ThingId.of("product", productId), metadataId));
    }

    @Override
    public int getOrder() {
        return Ordered.HIGHEST_PRECEDENCE + 100;
    }
}
